package ckR

//
//import (
//	"bytes"
//	"errors"
//	"fmt"
//	"db2s/dbExec"
//	ea "db2s/encryptionAlgorithm"
//	"db2s/global"
//	"db2s/outPut"
//	"db2s/scheduleTask"
//	"os"
//	"reflect"
//	"runtime"
//	"strings"
//	"sync"
//	"sync/atomic"
//)
//
//// blockingReadPipeData Blocking reading of dual pipe data
//func (sps SchedulePlan2) blockingReadPipeData(wg *sync.WaitGroup, pipe1, pipe2 chan any) (l chan any) {
//	l = make(chan any, GlobalPConfigs.repair.FixTrxNum)
//	go func() {
//		defer func() {
//			wg.Done()
//		}()
//		for {
//			v1, ok1 := <-pipe1
//			v2, ok2 := <-pipe2
//			if !ok1 && !ok2 {
//				close(l)
//				return
//			} else if ok1 && !ok2 {
//				l <- map[string]any{
//					"src": v1,
//				}
//			} else if !ok1 && ok2 {
//				l <- map[string]any{
//					"dst": v2,
//				}
//			} else if ok1 && ok2 {
//				l <- map[string]any{
//					"src": v1,
//					"dst": v2,
//				}
//			}
//		}
//	}()
//	return
//}
//
//// rowsMissSegmentQuery starts a query against segment
//func (sps SchedulePlan2) rowsMissSegmentQuery(wg *sync.WaitGroup, logThreadSeq int64, _ string) (sc, dc chan any, err error) {
//	var (
//		//event = "[rowsMissSegmentQuery]"
//		countThreshold int64 = 100
//		quitThreshold        = 2
//	)
//	sc = make(chan any, GlobalPConfigs.repair.FixTrxNum)
//	dc = make(chan any, GlobalPConfigs.repair.FixTrxNum)
//	taskObject := scheduleTask.GetTaskDistributionO(map[string]any{
//		"cpLength":         GlobalPConfigs.rules.QueueSize,
//		"whereColumn":      sps.indexColumn,
//		"schema":           sps.schema,
//		"table":            sps.table,
//		"logSeq":           logThreadSeq,
//		"tableLimit":       sps.tableLimit,
//		"chunkStartSource": sps.chunkStartSource,
//		"sdb":              GlobalPConfigs.SDB,
//		"ddb":              GlobalPConfigs.DDB,
//		"chunkStartFs":     sps.idxc,
//		"chunkStartFd":     sps.dstIdxc,
//		"chunkNumber":      GlobalPConfigs.rules.ChanRowCount,
//	})
//	sdr, err1 := taskObject.MissIndexColumQuery(logThreadSeq)
//	if err1 != nil {
//		return
//	}
//
//	go func(schema, table string, src, dst chan *string) {
//		defer wg.Done()
//		var (
//			//原目标端分割计数器
//			srcCounter, dstCounter int64
//			//原目标端数据分割
//			srcResultPointer, dstResultPointer []*string
//			//原目标端总数
//			srcRowsSum, dstRowsSum int64
//			//退出管道
//			quitDone                       = make(chan struct{}, 2)
//			srcCloseStatus, dstCloseStatus = false, false
//		)
//		for {
//			select {
//			case d, ok := <-dst:
//				if !ok {
//					if !dstCloseStatus {
//						//atomic.SwapInt64(sps.subTaskInfo.RecordCount, dstRowsSum)
//						//atomic.SwapInt64(terminalPods[getTableName(schema, table)].Rows.DestNumber, dstRowsSum)
//						if len(dstResultPointer) > 0 {
//							dc <- dstResultPointer
//						}
//						close(dc)
//						dstResultPointer = []*string{}
//						quitDone <- struct{}{}
//						dstCloseStatus = true
//					}
//				} else {
//					//计数
//					dstCounter++
//					dstRowsSum++
//					//终端实时计数
//					if PlanContext.TaskBarSubsStatusGet(sps.subTaskInfo) != -2 {
//						PlanContext.SubTaskBarAccumulate(map[string]any{"subTask": sps.subTaskInfo, "select": int64(0), "insert": int64(1)})
//					}
//					atomic.AddInt64(terminalPods[getTableName(schema, table)].Rows.DestNumber, 1)
//					//合并
//					if dstCounter == countThreshold {
//						dc <- dstResultPointer
//						dstResultPointer = []*string{}
//						dstCounter = 0
//					}
//					dstResultPointer = append(dstResultPointer, d)
//				}
//
//			case c, ok := <-src:
//				if !ok {
//					if !srcCloseStatus {
//						//atomic.SwapInt64(sps.subTaskInfo.RecordCount, srcRowsSum)
//						//atomic.SwapInt64(terminalPods[getTableName(schema, table)].Rows.SourceNumber, srcRowsSum)
//						if len(srcResultPointer) > 0 {
//							sc <- srcResultPointer
//						}
//						close(sc)
//						srcResultPointer = []*string{}
//						quitDone <- struct{}{}
//						srcCloseStatus = true
//					}
//				} else {
//					//计数
//					srcCounter++
//					srcRowsSum++
//					if PlanContext.TaskBarSubsStatusGet(sps.subTaskInfo) != -2 {
//						PlanContext.SubTaskBarAccumulate(map[string]any{"subTask": sps.subTaskInfo, "select": int64(1), "insert": int64(0)})
//					}
//					atomic.AddInt64(terminalPods[getTableName(schema, table)].Rows.SourceNumber, 1)
//					if srcCounter == GlobalPConfigs.rules.Sync.FixRows {
//						sc <- srcResultPointer
//						srcResultPointer = []*string{}
//						srcCounter = 0
//					}
//					srcResultPointer = append(srcResultPointer, c)
//				}
//			}
//			//错误中断机制
//			if e := atomic.LoadInt64(sps.errEnd); e < 0 {
//				var exitStatus int64 = -2
//				atomic.SwapInt64(sps.subTaskInfo.RecordCount, exitStatus)
//				err = errors.New(fmt.Sprintf("%s", "An error message occurred"))
//				atomic.AddInt64(sps.errEnd, -1)
//				if PlanContext.TaskBarSubsStatusGet(sps.subTaskInfo) != -2 {
//					PlanContext.TaskBarSubsStatusSwap(sps.subTaskInfo, "error")
//				}
//				//PlanContext.TpodTableStatusSwap(terminalPods[getTableName(schema, table)].Sync.SyncStatus, "error")
//				close(sc)
//				close(dc)
//				return
//			}
//			//退出机制
//			if len(quitDone) == quitThreshold {
//				if *terminalPods[getTableName(schema, table)].Rows.SourceNumber == 0 && *terminalPods[getTableName(schema, table)].Rows.DestNumber == 0 {
//					atomic.SwapInt64(sps.mqProductCustomerMonitor.mqStartTiming, 1)
//					atomic.SwapInt64(sps.mqProductCustomerMonitor.mqLevelProductSend, 1)
//					atomic.SwapInt64(sps.mqProductCustomerMonitor.mqLevelCustomerReceive, 1)
//				}
//				close(quitDone)
//				return
//			}
//		}
//	}(sps.schema, sps.table, sdr["src"], sdr["dst"])
//	return
//}
//
//// Md5Rows 针对表的所有列的数据类型进行处理，将列类型转换成字符串，例如时间类型，并执行sql语句
//func (sps *SchedulePlan2) rowsMissMd5Rows(stt, dtt []*string) (*string, *string, bool) {
//	var (
//		vlog             string
//		aa               = &ea.CheckSumTypeStruct{}
//		logseq           = 1
//		sstring, dstring string
//		buff             bytes.Buffer
//	)
//	for k, v := range stt {
//		if k+1 == len(stt) {
//			buff.WriteString(fmt.Sprintf("%v", *v))
//		} else {
//			buff.WriteString(fmt.Sprintf("%v%v", *v, global.DataRowsSplist))
//		}
//	}
//	sstring = buff.String()
//	buff.Reset()
//	for k, v := range dtt {
//		if k+1 == len(stt) {
//			buff.WriteString(fmt.Sprintf("%v", *v))
//		} else {
//			buff.WriteString(fmt.Sprintf("%v%v", *v, global.DataRowsSplist))
//		}
//	}
//	dstring = buff.String()
//	buff.Reset()
//	if aa.CheckMd5(sstring) != aa.CheckMd5(dstring) {
//		atomic.AddInt64(sps.abnormalDataAnalyze.abTotal, 1)
//		return &sstring, &dstring, false
//	} else {
//		runtime.SetFinalizer(stt, nil)
//		runtime.SetFinalizer(dtt, nil)
//		vlog = fmt.Sprintf("(%d) Verification table %s.%s The block data verified by the original target end is consistent.", logseq, sps.schema, sps.table)
//		global.Wlog.Debug(vlog)
//	}
//	vlog = fmt.Sprintf("(%d) The block data verification of check table %s.%s is completed !!!", logseq, sps.schema, sps.table)
//	global.Wlog.Debug(vlog)
//	return nil, nil, true
//}
//
//// rowsMissArrcmp 针对表的所有列的数据类型进行处理，将列类型转换成字符串，例如时间类型，并执行sql语句
//func (sps *SchedulePlan2) rowsMissArrcmp(stt, dtt *string, logseq int64) map[string][]*string {
//	var (
//		aa         = &ea.CheckSumTypeStruct{}
//		event      string
//		A          = make(map[string][]*string)
//		rowsSplict = func(s *string) (s1 []string) {
//			if strings.Contains(*s, string(RowsDataSplict)) {
//				s1 = strings.Split(*s, string(RowsDataSplict))
//			} else {
//				if len(*s) > 0 {
//					s1 = []string{*s}
//				}
//			}
//			return s1
//		}
//	)
//	global.Wlog.Info(fmt.Sprintf("(%d) [%s] Perform rows data arrcmp of table %s.%s.", logseq, event, sps.schema, sps.table))
//	add, del := aa.Arrcmp(rowsSplict(stt), rowsSplict(dtt))
//	if len(del) > 0 {
//		var del1 []*string
//		for _, v := range del {
//			v1 := v
//			del1 = append(del1, &v1)
//		}
//		A[string(Ldelete)] = del1
//	}
//	if len(add) > 0 {
//		var add1 []*string
//		for _, v := range add {
//			v1 := v
//			add1 = append(add1, &v1)
//		}
//		A[string(Linsert)] = add1
//	}
//	runtime.SetFinalizer(stt, nil)
//	runtime.SetFinalizer(dtt, nil)
//	global.Wlog.Debug(fmt.Sprintf("(%d) [%s] The rows data arrcmp of table %s.%s is completed.", logseq, event, sps.schema, sps.table))
//	return A
//}
//func (sps *SchedulePlan2) rowsMissAbnormalInsertDataPersistence(add []*string) (md5V []string, writeData map[string]*string, err error) {
//	var (
//		vlog string
//	)
//	if len(add) == 0 {
//		return
//	}
//	writeData = make(map[string]*string)
//	vlog = fmt.Sprintf("(%d) Start to generate the delete statement of check table %s.%s.", 1, sps.schema, sps.table)
//	global.Wlog.Debug(vlog)
//	for _, v := range add {
//		md5 := ea.CheckSum().CheckSha1(*v)
//		md5V = append(md5V, md5)
//		s := fmt.Sprintf("%v%v%v%v%v", "insert", global.DataColumnsSplist, md5, global.SchemaTableSplist, *v)
//		writeData[md5] = &s
//	}
//	vlog = fmt.Sprintf("(%d) The delete repair statement verifying table %s.%s is complete.", 1, sps.schema, sps.table)
//	global.Wlog.Debug(vlog)
//	return
//}
//func (sps *SchedulePlan2) rowsMissAbnormalDeleteDataPersistence(del []*string) (md5V []string, writeData map[string]*string, err error) {
//	var (
//		vlog string
//	)
//	if len(del) == 0 {
//		return
//	}
//	writeData = make(map[string]*string)
//	vlog = fmt.Sprintf("(%d) Start to generate the delete statement of check table %s.%s.", 1, sps.schema, sps.table)
//	global.Wlog.Debug(vlog)
//	for _, v := range del {
//		md5 := ea.CheckSum().CheckSha1(*v)
//		md5V = append(md5V, md5)
//		s := fmt.Sprintf("%v%v%v%v%v", "delete", global.DataColumnsSplist, md5, global.SchemaTableSplist, *v)
//		writeData[md5] = &s
//	}
//	vlog = fmt.Sprintf("(%d) The delete repair statement verifying table %s.%s is complete.", 1, sps.schema, sps.table)
//	global.Wlog.Debug(vlog)
//	return
//}
//
//// AbnormalDataDiff 差异数据的二次校验，并生成修复语句
//func (sps *SchedulePlan2) rowsMissAbnormalDataPersistence(diffQueryData map[string][]*string) error {
//	var (
//		vlog                   string
//		logThreadSeq           = 1
//		addMd5, delMd5         = make(map[string]string), make(map[string]string)
//		md5Uniq                = make(map[string]string)
//		addMd5Data, delMd5Data = make(map[string]*string), make(map[string]*string)
//		writeData, writeMd5    []*string
//	)
//	vlog = fmt.Sprintf("(%d) Check table %s.%s to start differential data processing and generate repair statements ...", logThreadSeq, sps.schema, sps.table)
//	global.Wlog.Info(vlog)
//	if del, ok := diffQueryData[string(Ldelete)]; ok {
//		if delS, md5Data, err := sps.rowsMissAbnormalDeleteDataPersistence(del); err != nil {
//			return err
//		} else {
//			for _, v := range delS {
//				delMd5[v] = "delete"
//			}
//			delMd5Data = md5Data
//		}
//	}
//	if add, ok := diffQueryData[string(Linsert)]; ok {
//		if addS, md5Data, err := sps.rowsMissAbnormalInsertDataPersistence(add); err != nil {
//			return err
//		} else {
//			for _, v := range addS {
//				addMd5[v] = "insert"
//			}
//			addMd5Data = md5Data
//		}
//	}
//	for k, v := range addMd5 {
//		if _, ok := delMd5[k]; !ok {
//			md5Uniq[k] = v
//		} else {
//			delete(delMd5, k)
//		}
//	}
//	for k, v := range delMd5 {
//		if _, ok := md5Uniq[k]; !ok {
//			md5Uniq[k] = v
//		}
//	}
//	for k := range md5Uniq {
//		k1 := k
//		writeMd5 = append(writeMd5, &k1)
//		if v, ok := delMd5Data[k]; ok {
//			writeData = append(writeData, v)
//		}
//		if v, ok := addMd5Data[k]; ok {
//			writeData = append(writeData, v)
//		}
//	}
//	WriteTextData := outPut.FileOut("text", map[string]any{"filen": sps.tmpFileF.tmpDataFileF, "logSeq": int64(1),
//		"bufSize": 1024 * 1024 * 4, "sqlType": "sql", "rowsS": global.DataRowsSplist,
//	})
//	if !WriteTextData.AppendWrite("", writeData) {
//		return errors.New(fmt.Sprintf(`write data fail!`))
//	}
//	WriteTextMd5 := outPut.FileOut("text", map[string]any{"filen": sps.tmpFileF.tmpMd5FileF, "logSeq": int64(1),
//		"bufSize": 1024 * 1024 * 4, "sqlType": "sql", "rowsS": global.DataRowsSplist,
//	})
//	if !WriteTextMd5.AppendWrite("", writeMd5) {
//		return errors.New(fmt.Sprintf(`write data fail!`))
//	}
//	vlog = fmt.Sprintf("(%d) Check table %s.%s to complete differential data processing and generate repair statements. !!!", logThreadSeq, sps.schema, sps.table)
//	global.Wlog.Info(vlog)
//	return nil
//}
//
//// rowsMissMd5ValueUniq 无索引差异数据md5全局唯一值去重
//func (sps *SchedulePlan2) rowsMissMd5ValueUniq() (md5UniqueConstraint map[string]int) {
//	var ()
//	md5UniqueConstraint = make(map[string]int)
//	ff := outPut.FileOut("text", map[string]any{"fileName": sps.tmpFileF.tmpMd5FileName, "rowsS": global.DataRowsSplist, "curry": 1})
//	readBuff := ff.Read().(chan *string)
//	for {
//		select {
//		case read, ok := <-readBuff:
//			if !ok {
//				return
//			}
//			if len(*read) == 0 {
//				continue
//			}
//			if _, ok1 := md5UniqueConstraint[*read]; !ok1 {
//				md5UniqueConstraint[*read]++
//			}
//		}
//	}
//}
//
//// rowsMissDataValueRead 根据md5唯一值读取差异数据文件，筛选数据差异
//func (sps *SchedulePlan2) rowsMissDataValueRead(md5UniqueConstraint map[string]int) chan any {
//	var (
//		resultC         = make(chan any, GlobalPConfigs.rules.QueueSize)
//		rowsDataSplistF = func(s string) (r RowsMissReadAbnormalDataFormat) {
//			if n := strings.Index(s, global.SchemaTableSplist); n != -1 {
//				if n1 := s[:n]; len(n1) > 0 {
//					if n2 := strings.Index(n1, global.DataColumnsSplist); n2 != -1 {
//						r.Action = n1[:n2]
//						r.Md5Value = n1[n2+len(global.DataColumnsSplist):]
//					}
//				}
//				if n1 := s[n+len(global.SchemaTableSplist):]; len(n1) > 0 {
//					r.RowsData = &n1
//				}
//			}
//			return
//		}
//	)
//	ff := outPut.FileOut("text", map[string]any{"fileName": sps.tmpFileF.tmpDataFileName, "rowsS": global.DataRowsSplist, "curry": 1})
//	readBuff := ff.Read().(chan *string)
//	go func() {
//		var rowsSum int64
//		var resultPointer []RowsMissReadAbnormalDataFormat
//		for {
//			select {
//			case read, ok := <-readBuff:
//				if !ok {
//					if len(resultPointer) > 0 {
//						resultC <- resultPointer
//						resultPointer = []RowsMissReadAbnormalDataFormat{}
//					}
//					close(resultC)
//					return
//				}
//				if len(*read) == 0 {
//					continue
//				}
//				l := rowsDataSplistF(*read)
//				if _, ok1 := md5UniqueConstraint[l.Md5Value]; ok1 {
//					//计数
//					rowsSum++
//					atomic.SwapInt64(sps.subTaskInfo.OutlierDetection, 1)
//					atomic.AddInt64(terminalPods[getTableName(sps.schema, sps.table)].Rows.RowsSum, 1)
//					atomic.AddInt64(sps.abnormalDataAnalyze.abTotal, 1)
//					if rowsSum == int64(GlobalPConfigs.repair.FixTrxNum) {
//						resultC <- resultPointer
//						resultPointer = []RowsMissReadAbnormalDataFormat{}
//						rowsSum = 0
//					}
//					resultPointer = append(resultPointer, l)
//					md5UniqueConstraint[l.Md5Value]++
//				}
//			}
//		}
//	}()
//	return resultC
//}
//
///*
//针对差异数据行做md5校验，并去除重复值
//*/
//func (sps *SchedulePlan2) rowsMissDataGeneralSql(chunkData any) (bool, int64, map[string][]*string, error) {
//	var (
//		insertColumn []string
//		//event        = "[loadDataInputGeneralInsert]"
//		insertPrefix string
//		//		chunkCount   []*string
//		lobSwitch  = false
//		GeneralSql = make(map[string][]*string)
//		insertSqlS []*string
//		deleteSqlS []*string
//	)
//	for _, v := range sps.tableColData.SColumnInfo {
//		insertColumn = append(insertColumn, v["columnName"])
//		if strings.Contains(v["dataType"], "lob") {
//			lobSwitch = true
//		}
//	}
//	sps.dbf.AppTable.LobSwitch = lobSwitch
//	sps.dbf.LogSeq = 1
//	if lobSwitch {
//		var s1 []string
//		for i := 0; i < len(insertColumn); i++ {
//			s1 = append(s1, "?")
//		}
//		s2 := dbExec.ApplyDataFix(sps.dbf).GeneralInsertPerfixSql(global.GeneralInsertPerfixSqlP{
//			Schema: sps.schema,
//			Table:  sps.table,
//			//PartitionName:    partitionName,
//			InsertColumn:     insertColumn,
//			PlaceholderValue: s1,
//		})
//		fmt.Println("----------aaaaa:", s2)
//		//chunkCount = append(chunkCount, &s2)
//	}
//	insertPrefix = dbExec.ApplyDataFix(sps.dbf).GeneralInsertPerfixSql(global.GeneralInsertPerfixSqlP{
//		Schema:       sps.schema,
//		Table:        sps.table,
//		InsertColumn: insertColumn,
//	})
//	switch fmt.Sprintf("%v", reflect.TypeOf(chunkData)) {
//	case "[]checkData.RowsMissReadAbnormalDataFormat":
//		for _, v := range chunkData.([]RowsMissReadAbnormalDataFormat) {
//			if strings.EqualFold(strings.TrimSpace(v.Action), "delete") {
//				atomic.AddInt64(terminalPods[getTableName(sps.schema, sps.table)].Rows.DeleteSum, 1)
//				atomic.AddInt64(sps.abnormalDataAnalyze.redundantTotal, 1)
//				if sqlstr, err1 := dbExec.ApplyDataFix(sps.dbf).FixDeleteSqlExec(*v.RowsData, sps.tableColData.SColumnInfo, "mui", sps.indexColumn, 1); err1 != nil {
//					//
//				} else {
//					sqlstr = fmt.Sprintf("%v\n", sqlstr)
//					deleteSqlS = append(deleteSqlS, &sqlstr)
//				}
//			} else if strings.EqualFold(strings.TrimSpace(v.Action), "insert") {
//				atomic.AddInt64(terminalPods[getTableName(sps.schema, sps.table)].Rows.InsertSum, 1)
//				if sqlstr, err1 := dbExec.ApplyDataFix(sps.dbf).FixInsertSqlExec(*v.RowsData, sps.tableColData.DColumnInfo, 1); err1 != nil {
//
//				} else {
//					atomic.AddInt64(sps.abnormalDataAnalyze.missTotal, 1)
//					sqlstr = fmt.Sprintf("%v (%v);\n", insertPrefix, strings.Join(strings.Split(sqlstr, global.DataColumnsSplist), ","))
//					insertSqlS = append(insertSqlS, &sqlstr)
//				}
//			}
//		}
//	}
//	GeneralSql["insert"] = insertSqlS
//	GeneralSql["delete"] = deleteSqlS
//	//if lobSwitch {
//	//	return true, int64(len(chunkCount)) - 1, GeneralSql, nil
//	//}
//	//				runtime.SetFinalizer(v, nil)
//	return lobSwitch, 0, GeneralSql, nil
//}
//
//// DataFixDiff 修复差异数据
//func (sps *SchedulePlan2) rowsMissDataFixDiff(fixSQL map[string][]*string, logThreadSeq int64) (err error) {
//	var (
//		wfile *os.File
//	)
//	for k, v := range fixSQL {
//		if strings.EqualFold(k, "insert") {
//			wfile = sps.repair.FixFileInsertF
//		} else if strings.EqualFold(k, "delete") {
//			wfile = sps.repair.FixFileDeleteF
//		}
//		sps.dbf.LogSeq = logThreadSeq
//		sps.dbf.AppTable.ShardName = sps.dshardName
//		sps.dbf.DataFixType = GlobalPConfigs.repair.DataFix
//		sps.dbf.AppTable.LobSwitch = false
//		sps.dbf.AppTable.AppType = k
//		sps.dbf.FullRowsc.WriteTextI = outPut.FileOut("text", map[string]any{"filen": wfile, "logSeq": logThreadSeq,
//			"bufSize": 1024 * 1024 * 4, "sqlType": "sql", "rowS": "\n",
//		})
//
//		if err = dbExec.ApplyDataFix(sps.dbf).ApplySql(global.ApplySqlP{
//			DB:             GlobalPConfigs.DDB,
//			ShardName:      sps.dshardName,
//			ApplySqlGather: v,
//			LogSeq:         logThreadSeq,
//		}); err != nil {
//			return err
//		}
//	}
//	return
//}
